package com.chukun.multi.producerconsumer.semaphore;


import com.chukun.multi.producerconsumer.ProductChannel;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;

/**
 * 基于 {@link Semaphore} 实现的对流量控制的队列
 * @author chukun
 * @param <P>
 */
public class SemaphoreBasedChannel<P> implements ProductChannel<P> {

    private final BlockingQueue<P> blockingQueue;
    private final Semaphore semaphore;

    /**
     *
     * @param blockingQueue 阻塞队列，通常是一个无界阻塞队列
     * @param flowLimit 流量限制数
     */
    public SemaphoreBasedChannel(BlockingQueue<P> blockingQueue,int flowLimit) {
        this.blockingQueue = blockingQueue;
        this.semaphore = new Semaphore(flowLimit);
    }
    @Override
    public P take() throws InterruptedException {
        return blockingQueue.take();
    }

    @Override
    public void put(P product) throws InterruptedException {
       semaphore.acquire();
       try{
           blockingQueue.put(product);
       }finally {
           semaphore.release();
       }
    }
}
